package net.flyinggroup.kafka.avro;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.UUID;

public class ConsumerChannelInterceptor implements ChannelInterceptor {
    /**
     * Invoked before the Message is actually sent to the channel.
     * This allows for modification of the Message if necessary.
     * If this method returns {@code null} then the actual
     * send invocation will not occur.
     *
     * @param message
     * @param channel
     */
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        Field field = ReflectionUtils.findField(MessageHeaders.class, "headers");
        field.setAccessible(true);
        @SuppressWarnings("unchecked")
        Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils.getField(field, message.getHeaders());
        headersMap.put(MessageHeaders.ID, UUID.nameUUIDFromBytes((byte[]) headersMap.get(MessageHeaders.ID)));
        headersMap.put("traceId", UUID.nameUUIDFromBytes((byte[]) headersMap.get("traceId")));
        return message;
    }
}
